package storm.bolt;

import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.util.Map;

public class OrderDetailSaveBolt extends BaseBasicBolt {
    private static JedisPool pool;
    private static Jedis jedis;
    public void prepare(Map conf, TopologyContext context){

    }
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        // 初始化Redis连接池
        pool = new JedisPool(new JedisPoolConfig(), "127.0.0.1");

        // 从连接池中获取一个连接
        jedis = pool.getResource();
        //获取订单总金额
        String total1 = tuple.getStringByField("total");
        total1 = total1.replace("total\":\"","");
        total1 = total1.replace("\"}],\"database\":\"bookorder","");
        Double total = Double.parseDouble(total1);
        //获取付款方式
        String methodofpayment = tuple.getStringByField("methodofpayment");
        methodofpayment = methodofpayment.replace("methodofpayment\":\"","");
        //获取下单的客户名
        String recipient = tuple.getStringByField("recipient");
        recipient = recipient.replace("recipient\":\"","");
        //获取下单的地址
        String receiveplace = tuple.getStringByField("receiveplace");
        receiveplace = receiveplace.replace("receiveplace\":\"","");
        receiveplace = receiveplace.substring(0,2);

        //System.out.println(recipient);
        //将客户名保存到有序集合set中，集合的属性为去重，重复的客户名加1
        jedis.zincrby("recipient",1,recipient);
        //每接受到一条元组将总销售额加上订单总金额total
        jedis.incrByFloat("totals",total);
        //将付款方式存入methodofpayment
        jedis.zincrby("method_of_payment",1,methodofpayment);
        //将order的book_name保存在bookname的有序集合中，每个bookname元素加a
        jedis.incrBy("ordernum",1);
        //计算下单客户数
        long a = jedis.zcard("recipient");
        //将客户数保存
        jedis.set("recipientnum",String.valueOf(a));
        //将下单地区的数量加1
        jedis.zincrby("receiveplaces",1,receiveplace);
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
